package org.example.task;

import com.google.gson.Gson;
import com.opencsv.exceptions.CsvValidationException;
import org.example.analyzer.DataAnalyzer;
import org.example.hbase.HBaseUtil;
import org.example.importer.DataImporter;
import org.example.web.websocket.ProgressSocket;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.function.Consumer;

public class DataRefreshTask implements Runnable {

    private final Consumer<String> onComplete;
    private final ProgressSocket progressSocket;
    private final Gson gson = new Gson();

    public DataRefreshTask(ProgressSocket progressSocket, Consumer<String> onComplete) {
        this.progressSocket = progressSocket;
        this.onComplete = onComplete;
    }

    @Override
    public void run() {
        try {
            // 步骤 1: 清空并重建HBase表
            broadcastProgress("log", "===== 步骤 1/3: 清空并重建HBase表... =====", 5);
            HBaseUtil.recreateTable();
            broadcastProgress("log", "HBase表已准备就绪。", 10);

            // 步骤 2: 导入数据
            broadcastProgress("log", "===== 步骤 2/3: 开始导入CSV数据... =====", 10);
            DataImporter dataImporter = new DataImporter();
            
            // 使用回调来获取并广播真实的导入进度
            // 数据导入过程占总进度的10%到80%
            dataImporter.importData("database.csv", (importPercentage) -> {
                int overallPercentage = 10 + (int)(importPercentage * 0.7); // 映射到10%-80%的区间
                broadcastProgress("log", "数据导入中... " + importPercentage + "%", overallPercentage);
            });
            broadcastProgress("log", "数据导入成功！", 80);


            // 步骤 3: 分析数据
            broadcastProgress("log", "===== 步骤 3/3: 分析数据... =====", 80);
            DataAnalyzer dataAnalyzer = new DataAnalyzer();
            String newJsonData = dataAnalyzer.analyze();
            broadcastProgress("log", "数据分析完成。", 95);

            // 任务完成
            broadcastProgress("log", "===== 所有步骤均已成功完成! =====", 100);
            onComplete.accept(newJsonData);

        } catch (IOException | CsvValidationException e) {
            broadcastProgress("error", "任务执行失败: " + e.getMessage(), -1);
        }
    }

    private void broadcastProgress(String type, String message, int percentage) {
        Map<String, Object> progressUpdate = new HashMap<>();
        progressUpdate.put("type", type);
        progressUpdate.put("message", message);
        progressUpdate.put("progress", percentage);
        progressSocket.broadcast(gson.toJson(progressUpdate));
    }
} 